package cn.doitedu.datayi.data_service.dao;

import cn.doitedu.datayi.data_service.utils.HbaseConnectionUtil;
import org.apache.commons.codec.digest.DigestUtils;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.filter.SubstringComparator;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.stereotype.Repository;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

@Repository
public class SessionQueryDaoImpl implements SessionQueryDao {
    @Override
    public List<String> query(Integer limit) throws IOException {
        Connection conn = HbaseConnectionUtil.getConn();
        Table table = conn.getTable(TableName.valueOf("t_session_aggr"));

        Scan scan = new Scan();
        scan.setLimit(limit);
        ResultScanner scanner = table.getScanner(scan);

        Iterator<Result> iterator = scanner.iterator();

        ArrayList<String> lst = new ArrayList<>();
        while(iterator.hasNext()){
            // 拿到一行
            Result result = iterator.next();
            byte[] jsonBytes = result.getValue("f".getBytes(), "q".getBytes());

            lst.add(new String(jsonBytes));
        }
        return lst;
    }


    @Override
    public List<String> queryByProvince(String province) throws IOException {
        Connection conn = HbaseConnectionUtil.getConn();
        Table table = conn.getTable(TableName.valueOf("t_session_aggr"));

        Scan scan = new Scan();
        String start_rowkey = DigestUtils.md5Hex(province);
        String end_rowkey = DigestUtils.md5Hex(province)+"z";

        byte[] bytes = Bytes.toBytes(Integer.MAX_VALUE);
        byte[] tmp = new byte[end_rowkey.getBytes().length+1];
        System.arraycopy(end_rowkey.getBytes(),0,tmp,0,end_rowkey.getBytes().length);
        tmp[tmp.length-1] = bytes[bytes.length-1];

        scan.withStartRow(start_rowkey.getBytes(),true);
        scan.withStopRow(tmp,true);

        ResultScanner scanner = table.getScanner(scan);

        Iterator<Result> iterator = scanner.iterator();

        ArrayList<String> lst = new ArrayList<>();
        while(iterator.hasNext()){
            // 拿到一行
            Result result = iterator.next();
            byte[] jsonBytes = result.getValue("f".getBytes(), "q".getBytes());

            lst.add(new String(jsonBytes));
        }
        return lst;
    }


    @Override
    public List<String> queryByCondition(String value) throws IOException {

        Connection conn = HbaseConnectionUtil.getConn();
        Table table = conn.getTable(TableName.valueOf("t_session_aggr"));

        Scan scan = new Scan();
        RowFilter rowFilter = new RowFilter(CompareOperator.EQUAL, new SubstringComparator(value));
        scan.setFilter(rowFilter);

        ResultScanner scanner = table.getScanner(scan);

        Iterator<Result> iterator = scanner.iterator();

        ArrayList<String> lst = new ArrayList<>();
        while(iterator.hasNext()){
            // 拿到一行
            Result result = iterator.next();
            byte[] jsonBytes = result.getValue("f".getBytes(), "q".getBytes());

            lst.add(new String(jsonBytes));
        }


        return lst;
    }


}
